package com.xujianlong.app.func;

import com.xujianlong.bean.TableProcess;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xujianlong.common.GmallConfig;
import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.sql.*;
import java.util.*;

public class TableProcessFunction extends BroadcastProcessFunction<JSONObject,String,JSONObject> {
    //HBASE侧输出流
    private OutputTag<JSONObject> outputTag;
    //Hbase连接
    private Connection connection;
    //广播map描述器
    private MapStateDescriptor<String,TableProcess> mapStateDescriptor;
    public TableProcessFunction(OutputTag<JSONObject> outputTag, MapStateDescriptor<String, TableProcess> mapStateDescriptor) {
        this.outputTag=outputTag;
        this.mapStateDescriptor=mapStateDescriptor;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //得到Phoenix连接
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        connection= DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);
    }
    //处理广播流
    @Override
    public void processBroadcastElement(String value, Context ctx, Collector<JSONObject> out) throws Exception {
        //得到的是mysql中的配置数据
        JSONObject jsonObject = JSONObject.parseObject(value);
        TableProcess tableProcess = JSON.parseObject(jsonObject.getString("after"), TableProcess.class);
        //如果输出类型是Hbase就需要创建表
        if(TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())){
            createTable(tableProcess.getSinkTable(),
                    tableProcess.getSinkPk(),
                    tableProcess.getSinkColumns(),
                    tableProcess.getSinkExtend());
        }
        //得到广播状态
        BroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key=tableProcess.getSourceTable()+"-"+tableProcess.getOperateType();
        //System.out.println("广播Key:"+key);
        //将状态广播出去
        broadcastState.put(key,tableProcess);

    }
    //拼接sql创建表
    //sql:create table db.tn (id varchar primary key,name varchar,sex varchar) ...;
    private void createTable(String sinkTable, String sinkPk, String sinkColumns, String sinkExtend) {

        if (sinkPk == null || sinkPk.equals("")) {
            sinkPk = "id";
        }

        if (sinkExtend == null) {
            sinkExtend = "";
        }

        //拼接建表SQL语句
        StringBuilder createSQL = new StringBuilder("create table if not exists ")
                .append(GmallConfig.HBASE_SCHEMA)
                .append(".")
                .append(sinkTable)
                .append("(");

        //切分字段
        String[] columns = sinkColumns.split(",");
        for (int i = 0; i < columns.length; i++) {
            String column = columns[i];
            if (sinkPk.equals(column)) {
                createSQL.append(column).append(" varchar primary key");
            } else {
                createSQL.append(column).append(" varchar");
            }

            //判断是否为最后一个字段
            if (i < columns.length - 1) {
                createSQL.append(",");
            }
        }

        createSQL.append(")").append(sinkExtend);

        //打印建表语句
        System.out.println("建表语句为：" + createSQL);

        PreparedStatement preparedStatement = null;

        try {
            //编译SQL
            preparedStatement = connection.prepareStatement(createSQL.toString());
            preparedStatement.execute();

        } catch (SQLException e) {
            e.printStackTrace();
            throw new RuntimeException("创建表：" + sinkTable + "失败！");
        } finally {
            if (preparedStatement != null) {
                try {
                    preparedStatement.close();
                } catch (SQLException e) {
                    e.printStackTrace();
                }
            }
        }
    }
    //处理主流
    @Override
    public void processElement(JSONObject value, ReadOnlyContext ctx, Collector<JSONObject> out) throws Exception {
        //拿到广播流状态，里面放着所有你自定义的，需要的数据
        ReadOnlyBroadcastState<String, TableProcess> broadcastState = ctx.getBroadcastState(mapStateDescriptor);
        String key = value.getString("tableName") + "-" + value.getString("type");
        TableProcess tableProcess = broadcastState.get(key);



        if(tableProcess!=null){
            //2.过滤字段
            filterColumn(value.getJSONObject("after"), tableProcess.getSinkColumns());
            value.put("sinkTable",tableProcess.getSinkTable());
            if(TableProcess.SINK_TYPE_HBASE.equals(tableProcess.getSinkType())){
                ctx.output(outputTag,value);
            }else if(TableProcess.SINK_TYPE_KAFKA.equals(tableProcess.getSinkType())){
                out.collect(value);
            }
        }else{
            System.out.println("组合Key:"+key+"不存在");
        }
    }
    private void filterColumn(JSONObject after, String sinkColumns) {
        String[] columns = sinkColumns.split(",");
        List<String> columnsList = Arrays.asList(columns);

        Set<Map.Entry<String, Object>> entries = after.entrySet();

        Iterator<Map.Entry<String, Object>> iterator = entries.iterator();
        while(iterator.hasNext()){
            if(!sinkColumns.contains(iterator.next().getKey())){
                iterator.remove();
            }
        }

    }


}
